package cn.dmrliu.edu.realtime.app.func;

import cn.dmrliu.edu.realtime.bean.TableProcess;
import cn.dmrliu.edu.realtime.common.EduConfig;
import cn.dmrliu.edu.realtime.util.PhoenixUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

/**
 * 主流和广播流的处理
 */
public class DimTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    // 关联广播状态
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private Map<String, TableProcess> tableProcessMap = new HashMap<>();

    public DimTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    // 初始化时为tableProcessMap赋值
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("com.mysql.cj.jdbc.Driver");

        String url = "jdbc:mysql://h102:3306/edu_config";

        Properties properties = new Properties();
        properties.put("user", "root");
        properties.put("password", "j)k*x~y0)*n_L)!o{y?C`w(1rV_viw");
        properties.put("useUnicode", "true");
        properties.put("characterEncoding", "utf8");
        properties.put("serverTimeZone", "Asia/Shanghai");
        properties.put("useSSL", "false");

        String sql = "select * from edu_config.table_process where sink_type = 'dim'";

        Connection connection = DriverManager.getConnection(url, properties);
        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        ResultSet resultSet = preparedStatement.executeQuery();
        ResultSetMetaData metaData = resultSet.getMetaData();

        while (resultSet.next()) {
            JSONObject jsonObject = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object columnValue = resultSet.getObject(i);
                jsonObject.put(columnName, columnValue);
            }
            TableProcess tableProcess = jsonObject.toJavaObject(TableProcess.class);
            tableProcessMap.put(tableProcess.getSourceTable(), tableProcess);
        }

        resultSet.close();
        preparedStatement.close();
        connection.close();
    }

    // 处理主流数据
    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext ctx, Collector<JSONObject> out) throws
            Exception {
        // TODO 1.获取表,算子状态
        String tableName = jsonObject.getString("table");
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        TableProcess tableProcess = null;

        // TODO 2.筛选维度信息处理
        if ((tableProcess = broadcastState.get(tableName)) != null
                || (tableProcess = tableProcessMap.get(tableName)) != null) {
            // 获取数据信息
            JSONObject dataJSONObject = jsonObject.getJSONObject("data");
            // 获得列信息
            String sinkColumns = tableProcess.getSinkColumns();
            // 移除不需要的列
            filterColumn(dataJSONObject, sinkColumns);
            // 获取写入的表信息
            String sinkTable = tableProcess.getSinkTable();
            // 补充信息
            dataJSONObject.put("sinkTable", sinkTable);
            out.collect(dataJSONObject);
        }
    }

    /**
     * 清除不用的列
     *
     * @param dataJSONObject
     * @param sinkColumns
     */
    private void filterColumn(JSONObject dataJSONObject, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJSONObject.entrySet();
        entrySet.removeIf(entry -> !columnList.contains(entry.getKey()));
    }

    // 处理广播流数据
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        // TODO 1.json字符串转为json对象
        JSONObject jsonObject = JSONObject.parseObject(jsonStr);

        // TODO 2.根据操作类型执行操作
        String op = jsonObject.getString("op");//获取操作符
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);//获取广播状态
        // 广播状态中保存配置信息，配置信息的sourceTable作为key
        if ("d".equals(op)) {
            TableProcess before = jsonObject.getObject("before", TableProcess.class);
            String sourceTable = before.getSourceTable();
            String sinkType = before.getSinkType();
            if ("dim".equals(sinkType)) {
                broadcastState.remove(sourceTable);
            }
        } else {
            TableProcess after = jsonObject.getObject("after", TableProcess.class);
            String sinkType = after.getSinkType();
            if ("dim".equals(sinkType)) {
                String sourceTable = after.getSourceTable();//key
                String sinkTable = after.getSinkTable();//写入的表
                String sinkColumns = after.getSinkColumns();//表字段
                String sinkPk = after.getSinkPk();//表主键
                String sinkExtend = after.getSinkExtend();//表扩展
                chaeckTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
                broadcastState.put(sourceTable, after);
            }
        }


    }

    // 创建维度表
    private void chaeckTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        if (sinkPk == null) {
            sinkPk = "id";
        }
        if (sinkExtend == null) {
            sinkExtend = "";
        }
        StringBuilder createSql = new StringBuilder();
        createSql.append("create table if not exists " + EduConfig.PHOENIX_SCHEMA + ".").append(sinkTable).append("(");

        String[] columns = sinkColumns.split(",");
        for (int i = 0; i < columns.length; i++) {
            String column = columns[i];
            if (column.equals(sinkPk)) {
                createSql.append(column).append(" varchar primary key ");
            } else {
                createSql.append(column).append(" varchar ");
            }
            if (i < columns.length - 1) {
                createSql.append(",");
            }
        }
        createSql.append(")").append(sinkExtend);
        System.out.println("phoenix中的建表语句是" + createSql);

        PhoenixUtil.executeSql(createSql.toString());
    }
}
